package com.horse.cloud.framework.redis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @create: Created by intelliJIDEA18.1.3
 * @author: bing.Pan
 * @e-mail: 15923508369@163.com
 * @gmdate: 24/07/2018 11:34 星期二
 * @sidesc: redis发布订阅配置
 */


@Configuration
public class RedisListenerConfig {

    @Autowired private TopicMessageListener     messageListener;
    @Autowired private RedisExecutorProperties  redisExecutorProperties;



    @Bean
    RedisMessageListenerContainer container(JedisConnectionFactory jedisConnectionFactory, Executor executor) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(jedisConnectionFactory);

        container.setTaskExecutor(executor);
        container.addMessageListener(messageListener, new PatternTopic("chat01"));
        container.addMessageListener(messageListener, new PatternTopic("_keyevent@0_:expired"));

        return container;
    }





    @Bean
    @Primary
    public Executor redisTaskAsyncPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(redisExecutorProperties.getCorePoolSize());
        executor.setMaxPoolSize(redisExecutorProperties.getMaxPoolSize());
        executor.setQueueCapacity(redisExecutorProperties.getQueueCapacity());
        executor.setKeepAliveSeconds(redisExecutorProperties.getKeepAliveSeconds());
        executor.setThreadNamePrefix(redisExecutorProperties.getThreadNamePrefix());

        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

}
